package com.idroc.coelacanth.service.servant;

import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;

import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.coelacanth.core.dht.ConsistendHashNode;
import com.idroc.coelacanth.core.dht.ConsistentHashRing;
import com.idroc.coelacanth.core.dht.NodeActionObserver;
import com.idroc.coelacanth.core.dht.NodeActionObserver.NodeAction;
import com.idroc.coelacanth.core.membership.Endpoint;
import com.idroc.coelacanth.core.message.MessageDispatchHandler;
import com.idroc.coelacanth.service.MessageService;
import com.idroc.coelacanth.service.ThreadPoolService;

/**
 * 存储服侍器
 * @author yuyoo (yuyoo4j@163.com)
 * @date 2009-12-18 下午06:20:09
 */
public final class StorageServant {
	
	private static final Logger log = LoggerFactory.getLogger(StorageServant.class);
	
	private ConsistendHashNode currentNode = null;
	
	private ConsistentHashRing ring = null;
	
	private final List<StorageClientBroker> registers = new LinkedList<StorageClientBroker>();
	
	private MessageService msgService = null;
	
	private final int heartbeatInterval = 5000;
	private final long heartbeatLifecycle = 30000; // 120s
	private final long pulseVersionBeginning = 0; // 0
	
	private final long pulseVersion = pulseVersionBeginning;

	public StorageServant(ConsistendHashNode node, MessageService msgService, ConsistentHashRing ring) {
		   
		this.currentNode = node;
		this.msgService = msgService;
		this.ring = ring;
		this.ring.addNodeActionObserver(new StorageNodeActionObserver());
		registerMessageHandler();
	}

	private void registerMessageHandler() {
		
		MessageDispatchHandler dispatcher = msgService.getServerMessageFactory().getMessageDispatcher();
		// 处理"存储服侍器集合请求消息"
		dispatcher.addUpMessageHandler(ServantSetRequestMessage.class, new ServantSetRequestMessageHandler());
		// 处理"客户端注册消息"
		dispatcher.addUpMessageHandler(ClientRegisterMessage.class, new ClientRegisterMessageHandler());
	}
	
	/**
	 * 节点发起动作, 通知已经注册到本机的存储客户端
	 * @param action -- 节点动作
	 * @param node -- 节点
	 * @param ring -- 一致性hash环
	 */
	private void noticeStorageClient(final NodeAction action, final int nodeNo, final ConsistentHashRing ring) {
		 
		for (StorageClientBroker item : registers) {
			try {
				final StorageClientBroker sc = item;
				Runnable task = new Runnable() {
					public void run() {
						sc.update(action, nodeNo, ring); 				
					}
				};
				ThreadPoolService.getMaintainThreadPool().submit(task);
			} catch (Exception ioex) {
				log.error("节点[{}]发起动作[{}]", nodeNo, action);
				log.error("通知节点动作观察者[{}]异常", item);
				log.error("通知异常", ioex);
			}
		}
	}
	
	/**
	 * 存储节点动作观察者
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-16 下午01:24:34
	 */
	private class StorageNodeActionObserver implements NodeActionObserver {
		
		private final Logger log = LoggerFactory.getLogger(this.getClass());

		public void update(NodeAction na, int nodeNo, ConsistentHashRing ring) { 
			
			ConsistendHashNode cn = ring.getCurrentNode();
			if (cn.getNo() == nodeNo) {
				log.debug("当前节点发生动作,通知观察者");
				updateSelfNode(na, nodeNo, ring);
			} else {
				log.debug("其他节点发生动作,通知观察者");
				updateOtherNode(na, nodeNo, ring);
			}
		} 
		
		/*
		 * 当前节点
		 */
		private void updateSelfNode(NodeAction na, int nodeNo, ConsistentHashRing ring) {
			
			if (na == NodeAction.INSERT) {
				ring.getOpens().put(Integer.valueOf(nodeNo), currentNode);
				ring.setCurrentNode(currentNode);
				log.debug("当前节点加入开放集合[{}], 并通知客户端", ring.getOpens()); 
				noticeStorageClient(na, nodeNo, ring); // 通知客户端
			} else {
				log.error("未知节点动作[{}]", na);
			} 
		}
		
		/*
		 * 其他节点
		 */
		private void updateOtherNode(NodeAction na, int nodeNo, ConsistentHashRing ring) {
 
			ConsistendHashNode current = ring.getCurrentNode();
			
			if (na == NodeAction.INSERT) {

				Integer nn = Integer.valueOf(nodeNo);
				ConsistendHashNode ib = ring.getSeeds().remove(nn);
				ring.getOpens().put(nn, ib);
				
				log.debug("通过节点编号[{}]定位候选节点[{}]", nn, ib);
				
				Iterator<ConsistendHashNode> it = ring.clockwiseNodeSequence(nodeNo, false);
				ConsistendHashNode next = it.next();
				if (next.getNo() == current.getNo() && current instanceof HashStorageNode) {
					log.debug("通知注册的客户端列表");
					noticeStorageClient(na, nodeNo, ring); // 通知客户端
					//TODO: 这里需要一个迁移任务
				} else {
					log.debug("do nothing");
				}
			} else {
				log.error("未知节点动作[{}]", na);
			} 
		}
	}
	
	/**
	 * 存储服侍器集合请求消息 处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-16 下午05:38:57
	 */
	private class ServantSetRequestMessageHandler extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {

			final ServantSetRequestMessage msg = (ServantSetRequestMessage)e.getMessage();
			log.debug("节点动作响应消息 处理器,消息:{}", msg); 
			  
			ServantSetResponseMessage res = new ServantSetResponseMessage();
			res.setFrom(ring.getCurrentNode().getEndpoint());
			
			SortedMap<Integer, ConsistendHashNode> opens = ring.getOpens();
			Map<Integer, Endpoint> servants = new HashMap<Integer, Endpoint>(opens.size());
			for (Integer key : opens.keySet()) {
				servants.put(key, opens.get(key).getEndpoint());
			}
			res.setServants(servants);
			
			ChannelFuture cf = msgService.sendMessage(msg.getFrom(), res);
			cf.addListener(new ChannelFutureListener() {
	
				public void operationComplete(ChannelFuture writeFuture)
						throws Exception { 
	
					if (!writeFuture.isSuccess()) {
						log.error("任务发送异常,将立即关闭端点通道", writeFuture.getCause());
						Iterator<StorageClientBroker> it = registers.iterator();
						while (it.hasNext()) {
							StorageClientBroker broker = it.next();
							if (broker.getRemote().equals(msg.getFrom())) {
								it.remove();
								log.debug("移除异常观察者[{}]", broker);
								break;
							} 
						}
					}  
				} 
			}); 
		}
	}
	
	/**
	 * 客户端注册消息 处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-17 上午10:26:44
	 */
	private class ClientRegisterMessageHandler extends  SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {

			ClientRegisterMessage msg = (ClientRegisterMessage)e.getMessage();
			log.debug("客户端注册消息 处理器,消息:{}", msg);  
			
			StorageClientBroker ob = new StorageClientBroker();
			ob.setServant(currentNode.getEndpoint());
			ob.setRemote(msg.getFrom());
			ob.setMsgService(msgService);
			registers.add(ob);
			log.debug("当前注册的客户端集合[{}]", registers);
		}
	}
}
